/**
 * Software License, Version 1.0
 * 
 * Copyright 2008 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgment:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgment may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or Granules, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called Granules, 
 * nor may Indiana University or Community Grids Lab or Granules appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 * GENERATED USING SOFTWARE.
 */
package cgl.granules.samples.iterative;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.Enumeration;

import cgl.granules.dataset.Dataset;
import cgl.granules.dataset.DatasetCollection;
import cgl.granules.dataset.DatasetException;
import cgl.granules.dataset.FileAccess;
import cgl.granules.dataset.StreamingAccess;
import cgl.granules.exception.MarshallingException;
import cgl.granules.operation.MapReduceBase;
import cgl.granules.operation.MapReduceException;
import cgl.granules.results.ResultPayload;
import cgl.granules.results.Results;
import cgl.granules.results.ResultsFactory;
import cgl.narada.event.NBEvent;

/**
 * An example demonstrating the <code>Map</code> role within a Word Count
 * operation.
 * 
 * @author Shrideep Pallickara
 * 
 */
public class StageTwo extends MapReduceBase {
	private ResultPayload resultPayload;
	private String datasetIdentifier = "FileDataset-ID-IterativeStageTwo";
	private DatasetCollection datasetCollection;
	private FileAccess access = null;
	private boolean initialized = false;

	public StageTwo() {
		setDomain("Generic");
		setVersionInformation(200);
		setOperationIdentifier("Iterative-Stage-Two");
		resultPayload = ResultPayload.getInstance();
	}

	public void execute() {

		try {
			if (!initialized) {
				initializeAccessToFiles();
				initialized = true;
			}

			StreamingAccess mapReduceDataset = getMapReduceDataset();

			while (mapReduceDataset.isDataAvailable()) {
				NBEvent nbEvent = mapReduceDataset.getStreamingData();
				processResultsFromMap(nbEvent);
			}

		} catch (DatasetException e) {
			e.printStackTrace();
		} catch (IOException e) {
			System.out.println("Problems converting bytes to longs");
			e.printStackTrace();
		} catch (MarshallingException e) {
			System.out.println("Problems Marshalling: " + e.getMessage());
			e.printStackTrace();
		} catch (MapReduceException e) {
			System.out.println("Problems writing Results: " + e.getMessage());
			e.printStackTrace();
		}

	}

	private void initializeAccessToFiles() throws DatasetException, IOException {
		datasetCollection = getDatasetCollection();
		
		if (!datasetCollection.hasDataset(datasetIdentifier)) {
			System.out.println("Unable to locate dataset: " + datasetIdentifier);
			return;
		}
		Dataset dataset = datasetCollection.getDataset(datasetIdentifier);

		if (dataset.getDatasetType() == Dataset.FILES) {
			access = (FileAccess) dataset;
		} else {
			System.out.println("Incorrect datatset: "
					+ dataset.getDatasetType() + " initialized. Returning ...");
			return;
		}

		if (access == null) {
			System.out.println("Unable to initialize the dataset");
			return;
		}

		Enumeration filenames = access.getFilenames();
		while (filenames.hasMoreElements()) {
			String filename = (String) filenames.nextElement();
			// readFromFile(filename);
		}
	}

	private void readFromFile(String filename) throws DatasetException,
			IOException {
		String line = "";
		BufferedReader in = new BufferedReader(new FileReader(filename), 65536);
		while ((line = in.readLine()) != null) {
			// parse(line);
		}
		in.close();
	}

	private synchronized void processResultsFromMap(NBEvent nbEvent)
			throws IOException, MarshallingException, MapReduceException {

		String synopsis = (String) nbEvent.getContentSynopsis();
		byte[] resultBytes = nbEvent.getContentPayload();
		Results results = ResultsFactory.getInstance().getResults(resultBytes);
		byte[] resultPayloadBytes = results.getResultPayload();

		ResultPayload resultPayload = ResultPayload.getInstance();

		if (results.getResultType() == ResultPayload.STRING) {
			String receivedString = resultPayload.getString(resultPayloadBytes);
			if (receivedString.equals("TERMINATE")) {
				setTeminationConditionReached();
				writeTermination(true);
				System.out.println("StageTwo: Terminate command from StageOne "
						+ "processed ");
			}
			return;
		}

		if (results.getResultType() != ResultPayload.INT) {
			System.out.println("StageOne: Incorrect ResultType received");
			return;
		}

		/**
		 * Remainder of processing assumes that the resultPayload type is of
		 * type ResultPayload.INT
		 */
		int value = resultPayload.getInt(resultPayloadBytes);

		setReceivedOutputFrom(results.getInstanceIdentifier());

		System.out.println("StageTwo: Received results (" + value + ") on ["
				+ synopsis + "] from MAP (" + results.getInstanceIdentifier()
				+ ")");

		if (!isAwaitingOutputs()) {
			resetOutputTracker();
			value -= 50;
			writeResult(value, false);
		}

	}

	private void writeResult(int value, boolean processingComplete)
			throws MarshallingException, MapReduceException {
		boolean completeResult = true;
		ResultPayload resultPayload = ResultPayload.getInstance();
		byte[] resultPayloadBytes = resultPayload.getBytes(value);

		Results results = createResults(completeResult, processingComplete);
		results.setResultPayload(resultPayloadBytes);
		results.setResultType(ResultPayload.INT);

		writeResults(results);

	}

	private void writeTermination(boolean processingComplete)
			throws MarshallingException, MapReduceException {
		boolean completeResult = true;
		String terminateString = "TERMINATE";
		ResultPayload resultPayload = ResultPayload.getInstance();
		byte[] resultPayloadBytes = resultPayload.getBytes(terminateString);

		Results results = createResults(completeResult, processingComplete);
		results.setResultPayload(resultPayloadBytes);
		results.setResultType(ResultPayload.STRING);

		writeResults(results);
	}
}
